package com.xushuda.yas.storage;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map.Entry;
import java.util.Set;
import java.util.SortedMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * 在普通文件系统上存储，格式为SSTable，原理类似LevelDb 在basDir目录下有多个sst和bf文件，还有一个log.seq文件 1.sst 1.bf 2..3..4..
 * 
 * @author xushuda
 *
 */
public class SSTStorageEngine extends PermSlice {

    private Logger logger = Logger.getLogger(SSTStorageEngine.class.getName());

    private String baseDir;

    private static String DIR_SEPERATOR = "/";

    private static final String SST_SUFFIX = ".sst";

    private static final String BLOOM_FILTER_SUFFIX = ".bf";

    private static final String LOG_VERSION_FILE_NAME = "log.seq";

    private static final int SEQ_SIZE = 8; // long is 8 bytes

    private static final int BLOOM_FILTER_SIZE = 1 << 20;

    // private static final int IO_BUFFER_SIZE = 1 << 20;
    private static final int DEFAULT_BLOCK_SIZE = 1 << 17; // 128k

    // 存储no和sst的map
    private ArrayList<SSTools> SSTList;

    // 最大sst个数
    private static final int MAX_SST_NUM = 4;

    private volatile boolean merging;

    // 当前最大的sst编号
    private int currentNo;

    // 记录已经持久化的行对应的日志的最大序列号
    private MappedByteBuffer seq;

    private void init() throws IOException {
        merging = false;

        File dir = new File(baseDir);
        if (!dir.exists()) {
            dir.mkdirs();
        }

        // 扫描目录初始化数据文件和内存索引
        File[] files = dir.listFiles();

        // 初始化list
        SSTList = new ArrayList<SSTools>();
        // for (int i = 0; i < (files.length - 1) / 2; i++) {
        // SSTList.add(null);
        // }

        Set<Integer> tmpSet = new HashSet<Integer>();
        for (File file : files) {
            // 文件名
            String fileName = file.getName().trim().toLowerCase();
            if (fileName.endsWith(SST_SUFFIX)) {
                // SSTable 数据文件
                int no = Integer.parseInt((String) fileName.subSequence(0, fileName.indexOf(SST_SUFFIX)));
                if (!tmpSet.contains(no)) {
                    tmpSet.add(no);
                    SSTools sst = new SSTools(no);
                    SSTList.add(sst);
                }
            }

            // bloomFilter文件
            // else if (fileName.endsWith(BLOOM_FILTER_SUFFIX)) {
            // // bloom filter 文件
            // int no = Integer.parseInt((String) fileName.subSequence(0, fileName.indexOf(BLOOM_FILTER_SUFFIX)));
            // if (!SSTList.contains(no)) {
            // SSTools sst = new SSTools(no);
            // SSTList.add(no, sst);
            // }
            // SSTList.get(no).fileNum++;
            // }

            // seq 文件
            // else if (fileName.equals(LOG_VERSION_FILE_NAME)) {
            // // 记录log序列号的数据
            // // seq = mapLogSeqFile(fileName, SEQ_SIZE);
            // }
        }

        // 将log文件映射到seqBuf上
        seq = mapSeqFile(LOG_VERSION_FILE_NAME, SEQ_SIZE);

        currentNo = SSTList.size();

        // 检查一遍正确性
        // for (SSTools sst : SSTList) {
        //
        // if (!sst.check()) {
        // throw new IOException("error found for sst");
        // }
        //
        // if (sst.no > currentNo) {
        // currentNo = sst.no;
        // }
        // }

    }

    // 将文件映射到seq
    private MappedByteBuffer mapSeqFile(String fileName, int size) throws IOException {
        RandomAccessFile randomAccessFile = new RandomAccessFile(baseDir + fileName, "rw");
        FileChannel fileChannel = randomAccessFile.getChannel();
        MappedByteBuffer bbf = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, size);
        randomAccessFile.close();
        fileChannel.close();
        return bbf;
    }

    /**
     * key-way merge
     */
    private void kwMerge() {
        // TODO 加入归并，collapse...
        // for(SSTools sst : SSTList){
        // //...
        // sst.fi();
        // }
    }

    public SSTStorageEngine(Comparator<String> cmp, String dir) throws IOException {
        super(cmp);

        baseDir = dir.endsWith(DIR_SEPERATOR) ? dir : dir + DIR_SEPERATOR;

        // 从文件系统初始化
        init();
    }

    @Override
    public void dump(SortedMap<String, SeqData> dataSet) throws IOException {

        if (dataSet.isEmpty()) {
            // 错误
            return;
        }
        // init sst
        SSTools sst = new SSTools(++currentNo, dataSet); // FIXME 这个No在合并后可能会出现冲突啊

        // if (!sst.pw()) {
        // logger.log(Level.WARNING, "prepare writing error");
        // }
        //
        // // write data
        // for (Entry<String, SeqData> row : dataSet.entrySet()) {
        // // 现在没有写入中断机制，不能恢复写入了一半的文件
        // // seqBf.rewind();
        // // seqBf.putLong(row.getValue().seq);
        // // bloom filter
        // // sst.add(row.getKey());
        //
        // // write
        // sst.write(new Line(row.getKey(), row.getValue().value, row.getValue().born, row.getValue().tomb));
        // }
        //
        // // 结束写入，flush ，释放资源
        // sst.finish();
        // // flush数据
        // // sst.writer.flush();

        // 更新seq
        seq.rewind();
        seq.putLong(sst.getMaxSeqNo());
        seq.force();

        // 在完全dump到磁盘之后，加入到map中
        SSTList.add(sst);

        // sst.minK = dataSet.firstKey();
        // sst.maxK = dataSet.lastKey();

        // // 检查sst正确性
        // if (!sst.check()) {
        // return;
        // }

        // 检测sst数量，启动归并
        if (SSTList.size() > MAX_SST_NUM) {
            if (!merging) {
                merging = true;
                new Thread(new Runnable() {

                    @Override
                    public void run() {
                        logger.info("start k-way merge");
                        kwMerge();
                        merging = false;
                    }
                }).start();
            }
        }

    }

    @Override
    public String find(String k, long currentSeqNo) {
        Line line = null;
        for (SSTools sst : SSTList) {
            // TODO 加入多视图版本控制
            if (null != (line = sst.find(k, currentSeqNo))) {
                return line.getValue();
            }
        }

        return null;
    }

    @Override
    public boolean contains(String k, long currentSeqNo) {
        for (SSTools sst : SSTList) {
            if (sst.contains(k)) {
                return true;
            }
        }
        return false;
    }

    /**
     * 外部方法不应该直接调用插入数据
     */
    @Override
    @Deprecated
    public void insert(String k, String v, long seqNo) {

        throw new RuntimeException("error invocation");

    }

    @Override
    public void remove(String k, long seqNo) {
        // TODO Auto-generated method stub

    }

    @Override
    public void update(String k, String v, long seqNo) {
        // TODO Auto-generated method stub

    }

    @Override
    public String[] findBetween(String k, long currentSeqNo) {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public long getMaxSeqNo() {
        seq.rewind();
        return seq.getLong();
    }

    // private enum IDX_STATE {
    // INIT(0), INC(1), OVR(2);
    //
    // public final int state;
    //
    // IDX_STATE(int state) {
    // this.state = state;
    // }
    // }

    // /**
    // * 从磁盘上恢复一个sst文件
    // *
    // * @param no
    // * @return
    // */
    // public SSTools sstReveal(int no) {
    // try {
    // return new SSTools(no, true);
    // } catch (IOException e) {
    // }
    // return null;
    // }

    // /**
    // * 创建一个新的sst文件
    // *
    // * @param no
    // * @return
    // */
    // public SSTools sstCreate(int no) {
    // try {
    // return new SSTools(no, false);
    // } catch (IOException e) {
    // }
    // return null;
    // }

    /**
     * 对sst的文件操作的类
     * 
     * @author xushuda
     *
     */
    class SSTools {
        private int no;
        private RandomAccessFile reader;
        private RandomAccessFile writer;
        private BFilter bloomFilter;

        private ConcurrentSkipListMap<String, IdxNode> idx;

        private long maxSeqNo;

        private int maxCachedBlocks;

        private LinkedHashMap<String, byte[]> cacheBlock = new LinkedHashMap<String, byte[]>();

        // private String minK;
        // private String maxK;

        /**
         * 从磁盘上创建SST，只需要传入编号即可
         * 
         * @param no
         * @throws IOException
         */
        public SSTools(int no) throws IOException {

            // 检查文件正确性
            check();

            this.no = no;

            init();

            // disk上会读出 一些基本数据,重建内存索引
            initIdxMf();

        }

        /**
         * 利用dump过程中创建SST，需要传入这个sst的编号和导入的数据
         * 
         * @param no
         * @param dataSet
         * @throws IOException
         */
        public SSTools(int no, SortedMap<String, SeqData> dataSet) throws IOException {

            this.no = no;

            init();
            // 指定block的大小
            blockSize = 0 == blockSize ? DEFAULT_BLOCK_SIZE : blockSize;

            logger.info("dataSet has " + dataSet.size() + " lines to be dump");
            long bg = System.currentTimeMillis();
            dngi(dataSet);
            logger.info("dump cost " + (System.currentTimeMillis() - bg) + " ms");

            // 1/15 block can be cached per sst
            maxCachedBlocks = (int) writer.length() / (blockSize * 15) + 1;
        }

        /**
         * 是否包含，没有调用pr，fi调用方自行保证
         * 
         * @param key
         * @return
         */
        public boolean contains(String key) {
            return bloomFilter.contains(key);
        }

        // 使用本节点的索引来获取文件
        /**
         * 查找某行，没有调用pr，fi调用方自行保证,不验证版本信息。
         * 
         * @param key
         * @param seqNo
         * @return
         * @throws IOException
         */
        public Line find(String key, long currentNo) {
            if (!contains(key)) {
                return null;
            }

            Entry<String, IdxNode> entry = idx.floorEntry(key);
            if (null == entry) {
                return null;
            }

            try {
                byte[] tmp = getDataByCache(entry.getValue());
                // 扫描数据
                int p = 0;
                while (p < tmp.length) {
                    Line line = new Line(tmp, p, true);
                    // TODO 版本控制,
                    if (cmp.compare(line.getKey(), key) == 0) {
                        return line;
                    }
                    p = line.nxtPos();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return null;
        }

        /**
         * finish 结束读写操作，flush 数据，释放资源 必须要显示地调用这个方法，否则资源无法得到释放
         */
        public void fi() {
            try {
                if (writer != null) {
                    writer.close();
                }
                if (reader != null) {
                    reader.close();
                }
            } catch (IOException e) {
            }
            writer = null;
            reader = null;
        }

        public long getMaxSeqNo() {
            return maxSeqNo;
        }

        /**
         * 由某个索引节点从LRU缓存中获取已经读出来的数据
         * 
         * @param idxNode
         * @return
         * @throws IOException
         */
        private byte[] getDataByCache(IdxNode idxNode) throws IOException {

            if (!cacheBlock.containsKey(idxNode.getKey())) {
                int size = idxNode.getSize();
                long offset = idxNode.getOffset();

                byte[] tmp = new byte[size];
                // 读取数据
                reader.seek(offset);
                reader.readFully(tmp);
                if (cacheBlock.size() + 1 > maxCachedBlocks) {
                    Iterator<Entry<String, byte[]>> it = cacheBlock.entrySet().iterator();
                    // cacheBlock.remove(it.next().getKey());
                    it.next();
                    it.remove();
                }
                cacheBlock.put(idxNode.getKey(), tmp);
            }
            return cacheBlock.get(idxNode.getKey());
        }

        /**
         * (dump and generate index)(the index will also be dumped to disk after data) 持久化数据文件，索引文件，构建内存索引，
         * dump到硬盘上，写入数据
         * 
         * @param dataSet
         * @throws IOException
         */
        private void dngi(SortedMap<String, SeqData> dataSet) throws IOException {

            long currentOffset = 0; // 偏移量
            int currentSize = 0; // 字节数
            IdxNode tmpIdxNode = new IdxNode(currentOffset, dataSet.firstKey());
            idx = new ConcurrentSkipListMap<String, SSTStorageEngine.SSTools.IdxNode>(cmp);

            // -------------写入kv数据开始,在内存中初始化了索引-------------
            for (Entry<String, SeqData> d : dataSet.entrySet()) {
                SeqData data = d.getValue();
                String key = d.getKey();
                Line line = new Line(key, data.value, data.born, data.tomb);

                if (data.born > maxSeqNo) {
                    maxSeqNo = data.born;
                }

                // 可能当超过block的大小时,便生成新的索引节点
                if (currentSize + line.length() > blockSize) {

                    // 完成索引节点加入idx
                    tmpIdxNode.setSize(currentSize);
                    idx.put(tmpIdxNode.getKey(), tmpIdxNode);

                    // 生成新的索引节点
                    tmpIdxNode = new IdxNode(currentOffset, key);
                    currentSize = 0;
                }

                // 数据直接写入
                bloomFilter.add(line.getKey());

                writer.write(line.toBytes());

                // offset一直在增长
                currentOffset += line.length();
                currentSize += line.length();
            }

            // 最后一个节点的索引加入
            if (tmpIdxNode.getOffset() != currentOffset) {
                tmpIdxNode.setSize((int) (currentOffset - tmpIdxNode.getOffset()));
                idx.put(tmpIdxNode.getKey(), tmpIdxNode);
            }
            // -------------写入kv数据结束-------------

            // -------------写入索引数据开始-------------
            int idxSize = 0;
            for (Entry<String, IdxNode> d : idx.entrySet()) {
                tmpIdxNode = d.getValue();
                idxSize += tmpIdxNode.length();
                writer.write(tmpIdxNode.toBytes());
            }
            // -------------写入索引数据结束-------------

            // -------------写入manifest------------
            Manifest mf = new Manifest(idxSize, blockSize);
            writer.write(mf.toBytes());
            // -------------写入manifest结束---------

        }

        /**
         * 从磁盘上初始化manifest文件，重建内存中的索引
         * 
         * @throws IOException
         */
        private void initIdxMf() throws IOException {
            long rPos = reader.length();
            // 初始化blockSize 最后四个字节是block的大小，倒数8-4个字节是索引文件的大小
            reader.seek(rPos -= 8);
            byte[] b = new byte[8];
            reader.readFully(b);
            Manifest mf = new Manifest(b, 0);
            int idxSize = mf.getIdxSize();
            blockSize = mf.getBlockSize();

            // 读取索引文件内容
            byte[] idxBlock = new byte[idxSize];
            reader.seek(rPos -= idxSize);
            reader.readFully(idxBlock);

            // 初始化索引文件
            idx = new ConcurrentSkipListMap<String, SSTStorageEngine.SSTools.IdxNode>(cmp);

            // 将数据读入SkipList中
            int pos = 0;
            while (pos < idxSize) {
                IdxNode idxN = new IdxNode(idxBlock, pos, true);
                idx.put(idxN.getKey(), idxN);
                pos = idxN.getNextPos();
            }

            // 1/15 block can be cached per sst
            maxCachedBlocks = (int) writer.length() / (blockSize * 15) + 1;
        }

        /**
         * 在从磁盘初始化的时候检查
         * 
         * @throws RuntimeException 如果没有通过检测 程序异常，在这种情况下，文件部分写入，在下次打开的时候需要修复文件 重新定义这个为捕获异常这个异常应该由创建sst的调用放捕获，并且尝试恢复这块sst
         */
        private void check() throws IOException {

            // 要么两个都不存在，要么两个都存在
            if (new File(getSstPath()).exists() ^ new File(getBFPath()).exists()) {
                logger.log(Level.SEVERE, "error, check failure");
                // return false;
                throw new IOException("file missing when initializing SST");
            }
            // return true;

            /**
             * 该特性与dump方法关联。 TODO 灾难处理 容错
             * 
             * @see SSTStorageEngine
             */
        }

        private void init() throws IOException {
            if (writer == null) {
                createBW();
            }
            if (reader == null) {
                createBR();
            }
            if (bloomFilter == null) {
                createBF();
            }
        }

        private BFilter createBF() throws IOException {
            bloomFilter = new BFilter(getBFPath(), BLOOM_FILTER_SIZE);
            return bloomFilter;
        }

        private RandomAccessFile createBR() throws FileNotFoundException {
            reader = new RandomAccessFile(new File(getSstPath()), "r");
            return reader;
        }

        private RandomAccessFile createBW() throws IOException {
            writer = new RandomAccessFile(new File(getSstPath()), "rw");
            return writer;
        }

        private String getSstPath() {
            return baseDir + no + SST_SUFFIX;
        }

        private String getBFPath() {
            return baseDir + no + BLOOM_FILTER_SUFFIX;
        }

        /**
         * 
         * @author xushuda
         *
         */
        class IdxNode {
            long offset; // 启始地址

            int keyLen;

            int size; // 对应那个块的大小

            byte[] key;

            final int pos;

            public long getOffset() {
                return offset;
            }

            public int getSize() {
                return size;
            }

            public IdxNode(byte[] bb, int pos, boolean continuous) {
                this.pos = pos;
                offset = InnerUtils.BTL(bb, pos);
                keyLen = InnerUtils.BTI(bb, pos += 8);
                size = InnerUtils.BTI(bb, pos += 4);
                if (continuous) {
                    fillKey(bb, pos += 4);
                }
            }

            /**
             * 初始化不带size的IdxNode，后续需要调用setSize
             * 
             * @param offset
             * @param key
             */
            public IdxNode(long offset, String key) {
                this.offset = offset;
                pos = 0;
                this.key = key.getBytes();
                this.keyLen = this.key.length;
            }

            public void setSize(int size) {
                this.size = size;
            }

            /**
             * 填充key数据
             * 
             * @param bb
             * @param pos
             */
            public void fillKey(byte[] bb, int pos) {
                key = new byte[keyLen];
                System.arraycopy(bb, pos, key, 0, keyLen);
            }

            /**
             * 下一行完整数据的位置
             * 
             * @return
             */
            public int getNextPos() {
                return pos + 16 + keyLen;
            }

            /**
             * 
             * @return
             */
            public String getKey() {
                return new String(key);
            }

            /**
             * 返回byte数组
             * 
             * @return
             */
            public byte[] toBytes() {
                byte[] tmp = new byte[16 + keyLen];
                int p = 0;
                System.arraycopy(InnerUtils.LTB(offset), 0, tmp, p, 8);
                System.arraycopy(InnerUtils.ITB(keyLen), 0, tmp, p += 8, 4);
                System.arraycopy(InnerUtils.ITB(size), 0, tmp, p += 4, 4);
                System.arraycopy(key, 0, tmp, p += 4, keyLen);
                return tmp;
            }

            public int length() {
                return 16 + keyLen;
            }
        }

        class Manifest {
            // 索引大小
            int idxSize;
            // 文件的块大小
            int blockSize;

            public Manifest(byte[] b, int pos) {
                idxSize = InnerUtils.BTI(b, 0);
                blockSize = InnerUtils.BTI(b, 4);
            }

            public Manifest(int idxSize, int blockSize) {
                this.idxSize = idxSize;
                this.blockSize = blockSize;
            }

            public int length() {
                return 8;
            }

            public byte[] toBytes() {
                byte[] tmp = new byte[8];
                int p = 0;
                System.arraycopy(InnerUtils.ITB(idxSize), 0, tmp, p, 4);
                System.arraycopy(InnerUtils.ITB(blockSize), 0, tmp, p += 4, 4);
                return tmp;
            }

            public String toString() {
                return new String(toBytes());
            }

            public int getIdxSize() {
                return idxSize;
            }

            public int getBlockSize() {
                return blockSize;
            }

            /**
             * 不应该调用这个方法
             * 
             * @return
             */
            public int getNxtPos() {
                throw new RuntimeException("eorror manifest is read from file's tail ");
            }
        }
    }

    /**
     * 前面24字节为定长 字符串拼接顺序 fromKey+toKey+keyLen+valueLen+key+value 长度 (8bytes)(8bytes)(4bytes)(4bytes)(keyLen
     * bytes)(valueLen bytes) 所有字符串作为入参的方法都是deprecated，因为getBytes存在着创建内存空间与拷贝,另外主要是直接操作string不符合当前对文件的读取写入方式
     * 所有string为平台默认编码方式编码
     * 
     * @author xushuda
     *
     */
    static class Line {

        // 采用定长数据头的方式，不使用分割符
        // private static final String FIELD_DELIMITER = "\t";
        // private static final String ROW_DELIMITER = "\n";

        private byte[] key;
        private byte[] value;

        // 定长数据头
        public static final int HEAD_LEN = 24;
        // 开始与结束版本号
        private long fromSeq;
        private long toSeq;

        // key的长度和value的长度(多少byte)
        private int keyLen;
        private int valueLen;

        //
        private final int pos;

        /**
         * Deprecated，读出的时候#toString()用的默认的字符编码，这里不清楚传入参数是否用了别的字符编码? 并且这里在getBytes方法中进行了拷贝，效率比较低下
         * 
         * @param raw
         */
        @Deprecated
        public Line(String raw) {
            pos = 0;
            if (raw.getBytes().length != 24) {
                throw new RuntimeException("error initilization of sst line from string");
            }
            int p = 0;
            // 定长数据
            fromSeq = InnerUtils.BTL(raw.substring(0, p += 8).getBytes(), 0);
            toSeq = InnerUtils.BTL(raw.substring(p, p += 8).getBytes(), 0);
            keyLen = InnerUtils.BTI(raw.substring(p, p += 4).getBytes(), 0);
            valueLen = InnerUtils.BTI(raw.substring(p, p += 4).getBytes(), 0);
        }

        // public Line(byte[] bb, int pos) {
        //
        // this.pos = pos;
        // // 定长数据
        // // byte[] tmp = new byte[8];
        // // int q = 0;
        // // System.arraycopy(bb, q, tmp, 0, 8);
        // // fromSeq = BTL(tmp);
        // // System.arraycopy(bb, q += 8, tmp, 0, 8);
        // // toSeq = BTL(tmp);
        // // System.arraycopy(bb, q += 8, tmp, 0, 4);
        // // keyLen = BTI(tmp);
        // // System.arraycopy(bb, q += 4, tmp, 0, 4);
        // // valueLen = BTI(tmp);
        //
        // assert (bb.length >= pos + 24);
        // // 零拷贝实现
        // fromSeq = InnerUtils.BTL(bb, pos);
        // toSeq = InnerUtils.BTL(bb, pos += 8);
        // keyLen = InnerUtils.BTI(bb, pos += 8);
        // valueLen = InnerUtils.BTI(bb, pos += 4);
        // }

        /**
         * true 连续读取，直接读出key和value false 由从磁盘读取到的byte[]解析,只解析前24字节的定长数据 后续获取长度再读取相应字节从而获取key和value
         * 
         * @param bb
         * @param pos
         * @param continuous
         */
        public Line(byte[] bb, int pos, boolean continuous) {
            this.pos = pos;
            fromSeq = InnerUtils.BTL(bb, pos);
            toSeq = InnerUtils.BTL(bb, pos += 8);
            keyLen = InnerUtils.BTI(bb, pos += 8);
            valueLen = InnerUtils.BTI(bb, pos += 4);
            if (continuous) {
                fill(bb, pos += 4);
            }
        }

        /**
         * 由插入的实际数据创建SST
         * 
         * @param key
         * @param value
         * @param fromSeq
         */
        public Line(String key, String value, long fromSeq, long toSeq) {
            pos = 0;
            this.key = key.getBytes();
            this.value = value.getBytes();
            this.fromSeq = fromSeq;
            this.toSeq = toSeq;
            keyLen = this.key.length;
            valueLen = this.value.length;
        }

        /**
         * 获取下一个字段的开始位置
         * 
         * @return
         */
        public int nxtPos() {
            return pos + 24 + keyLen + valueLen;
        }

        /**
         * 此行数据的长度
         * 
         * @return
         */
        public int length() {
            return 24 + keyLen + valueLen;
        }

        /**
         * 获取key的长度
         * 
         * @return
         */
        public int getKeyLen() {
            return keyLen;
        }

        /**
         * 获取value的长度
         * 
         * @return
         */
        public int getValueLen() {
            return valueLen;
        }

        /**
         * 获取创建的序列号
         * 
         * @return
         */
        public long getBorn() {
            return fromSeq;
        }

        /**
         * 获取结束的序列号
         * 
         * @return
         */
        public long getTomb() {
            return toSeq;
        }

        /**
         * 获取key所表示的字符串
         * 
         * @return
         */
        public String getKey() {
            return new String(key);
        }

        /**
         * 获取value所表示的字符串
         * 
         * @return
         */
        public String getValue() {
            return new String(value);
        }

        /**
         * 读取出key
         * 
         * @param bb
         */
        public void fillKey(byte[] bb, int pos) {
            assert (bb.length >= pos + keyLen);
            key = new byte[keyLen];
            System.arraycopy(bb, pos, key, 0, keyLen);
        }

        /**
         * 读取出value
         * 
         * @param bb
         */
        public void fillValue(byte[] bb, int pos) {
            assert (bb.length >= pos + valueLen);
            value = new byte[valueLen];
            System.arraycopy(bb, pos, value, 0, valueLen);
        }

        /**
         * 直接填充所有数据
         * 
         * @param bb
         */
        public void fill(byte[] bb, int pos) {
            assert (bb.length >= pos + keyLen + valueLen);
            byte[] keyTmp = new byte[keyLen];
            byte[] valueTmp = new byte[valueLen];
            System.arraycopy(bb, pos, keyTmp, 0, keyLen);
            key = keyTmp;
            System.arraycopy(bb, pos + keyLen, valueTmp, 0, valueLen);
            value = valueTmp;
        }

        /**
         * 
         * @param key
         */
        @Deprecated
        public void fillKey(String key) {
            this.key = key.getBytes();
        }

        /**
         * 
         * @param value
         */
        @Deprecated
        public void fillValue(String value) {
            this.value = value.getBytes();
        }

        /**
         * 
         * @param knv
         */
        @Deprecated
        public void fill(String knv) {
            fill(knv.getBytes(), 0);
        }

        /**
         * 返回String的字符串
         */
        public String toString() {
            // StringBuilder sb = new StringBuilder();
            //
            // sb.append(LTB(fromSeq)).append(LTB(toSeq)).append(ITB(keyLen)).append(ITB(valueLen))
            // .append(new String(key)).append(new String(value));
            //
            // return sb.toString();
            return new String(toBytes());
        }

        /**
         * 生成新的byte[]
         * 
         * @return
         */
        public byte[] toBytes() {
            byte[] ret = new byte[24 + keyLen + valueLen];
            int q = 0;
            System.arraycopy(InnerUtils.LTB(fromSeq), 0, ret, q, 8);
            System.arraycopy(InnerUtils.LTB(toSeq), 0, ret, q += 8, 8);
            System.arraycopy(InnerUtils.ITB(keyLen), 0, ret, q += 8, 4);
            System.arraycopy(InnerUtils.ITB(valueLen), 0, ret, q += 4, 4);
            System.arraycopy(key, 0, ret, q += 4, keyLen);
            System.arraycopy(value, 0, ret, q += keyLen, valueLen);
            return ret;
        }

    }

    /**
     * 公用工具类
     * 
     * @author xushuda
     *
     */
    static class InnerUtils {
        /**
         * long转化为string
         * 
         * @param x
         * @return
         */
        public static byte[] LTB(long x) {
            byte[] bb = new byte[8];
            bb[0] = (byte) (x >> 56);
            bb[1] = (byte) (x >> 48);
            bb[2] = (byte) (x >> 40);
            bb[3] = (byte) (x >> 32);
            bb[4] = (byte) (x >> 24);
            bb[5] = (byte) (x >> 16);
            bb[6] = (byte) (x >> 8);
            bb[7] = (byte) (x >> 0);
            // 使用platform 的默认字符集编码，解码的时候使用相同的字符集，所以不会有问题
            return bb;
        }

        /**
         * int to byte
         * 
         * @param x
         * @return
         */
        public static byte[] ITB(int x) {
            byte[] bb = new byte[8];
            bb[0] = (byte) (x >> 24);
            bb[1] = (byte) (x >> 16);
            bb[2] = (byte) (x >> 8);
            bb[3] = (byte) (x >> 0);
            return bb;
        }

        /**
         * BTI byte to int
         * 
         * @param bb
         * @param pos
         * @return
         */
        public static int BTI(byte[] bb, int pos) {
            return ((((int) bb[pos] & 0xff) << 56) | (((int) bb[pos + 1] & 0xff) << 48)
                    | (((int) bb[pos + 2] & 0xff) << 40) | (((int) bb[pos + 3] & 0xff) << 32));
        }

        /**
         * string 转化为 long
         * 
         * @param data
         * @return
         */
        public static long BTL(byte[] bb, int pos) {
            return ((((long) bb[pos] & 0xff) << 56) | (((long) bb[pos + 1] & 0xff) << 48)
                    | (((long) bb[pos + 2] & 0xff) << 40) | (((long) bb[pos + 3] & 0xff) << 32)
                    | (((long) bb[pos + 4] & 0xff) << 24) | (((long) bb[pos + 5] & 0xff) << 16)
                    | (((long) bb[pos + 6] & 0xff) << 8) | (((long) bb[pos + 7] & 0xff) << 0));
        }
    }

    public static void main(String[] args) {
        Line x = new Line("1位", "2a", 3, 4);
        Line y = new Line(x.toBytes(), 0, true);
        // y.fillKey(x.toBytes(), 24);
        // y.fillValue(x.toBytes(), 24 + x.getKeyLen());
        if (y.fromSeq == 3 && y.toSeq == 4 && y.getKey().equals("1位") && y.getValue().equals("2a")) {
            System.out.println(x.toString());
            System.out.println("OK!");
        } else {
            System.out.println("ERROR!");
        }
    }
}
